package org.acme.service;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;


@Service
public class TestServiceImpl implements InitializingBean, DisposableBean, Runnable {

    Logger logger = LoggerFactory.getLogger(getClass());

    private Thread thread;

    private BinaryLogClient binaryLogClient;

    @Override
    public void destroy() throws Exception {
        if (this.binaryLogClient.isConnected()) {
            this.binaryLogClient.disconnect();
        }
        if (this.thread.isAlive()) {
            this.thread.stop();
        }

    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.thread = new Thread(this::run);
        this.thread.setDaemon(true);
        this.thread.setName("dbListener");
        this.thread.start();
    }

    @Override
    public void run() {
        try {
            binaryLogClient = new BinaryLogClient("127.0.0.1", 3306, "canal", "root", "My7Pass@Word_9_8A_zE");
            binaryLogClient.setEventDeserializer(new EventDeserializer());
            binaryLogClient.registerLifecycleListener(new BinaryLogClient.LifecycleListener() {
                @Override
                public void onConnect(BinaryLogClient client) {
                    logger.warn("binaryLogClient onConnect ...");
                }

                @Override
                public void onCommunicationFailure(BinaryLogClient client, Exception ex) {

                }

                @Override
                public void onEventDeserializationFailure(BinaryLogClient client, Exception ex) {

                }

                @Override
                public void onDisconnect(BinaryLogClient client) {
                    logger.warn("binaryLogClient onDisconnect ...");
                }
            });
            binaryLogClient.registerEventListener(new BinaryLogClient.EventListener() {
                @Override
                public void onEvent(Event event) {
                    EventHeader header = event.getHeader();
                    EventType eventType = header.getEventType();
                    if (event.getData() instanceof TableMapEventData) {
                        logger.info(event.getData().toString());
                    }
                    if (EventType.isWrite(eventType)) {
                        WriteRowsEventData data = event.getData();
                        logger.info(data.toString());
                    } else if (EventType.isUpdate(eventType)) {
                        UpdateRowsEventData data = event.getData();
                        logger.info(data.toString());
                    } else if (EventType.isDelete(eventType)) {
                        DeleteRowsEventData data = event.getData();
                        logger.info(data.toString());
                    }
                }
            });
            binaryLogClient.connect();
        } catch (Exception e) {
            logger.error("connect error :" + e.getMessage());
        }
    }
}
